package com.leilei.count;

import com.leilei.Location;
import com.leilei.sink.RabbitMqSinkProperties;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/14 21:55
 * @desc flink 窗口 根据数量开窗口
 */
public class FlinkWindow_CountNoKeyWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(4);
        DataStreamSource<Location> source = env.addSource(new MyLocationSource());
        /**
         * 无key窗口，表示你的数据流不会被分成多条流执行，所以的窗口都会在同一个task中计算，也就是并行度parallesim 为1。
         */
        AllWindowedStream<Location, GlobalWindow> noKeyWindow = source.countWindowAll(2);
        // //内部类形式
        // // noKeyWindow.apply(new AllWindowFunction<Location, Object, GlobalWindow>() {
        // //     @Override
        // //     public void apply(GlobalWindow window, Iterable<Location> values, Collector<Object> out) throws Exception {
        // //        //todo
        // //     }
        // // });
        SingleOutputStreamOperator<String> noKeyWindowResult = noKeyWindow.apply(new CountNoKeyWindowFunction());
        noKeyWindowResult.print();
        //String path = "RabbitMqSink.properties";
        //Props props = new Props(path);
        //RabbitMqSinkProperties sinkProperties = RabbitMqSinkProperties.builder()
        //        .host(props.getStr("sink.rabbitmq.host"))
        //        .port(props.getInt("sink.rabbitmq.port"))
        //        .userName(props.getStr("sink.rabbitmq.username"))
        //        .passWord(props.getStr("sink.rabbitmq.password"))
        //        .exchange(props.getStr("sink.rabbitmq.exchange"))
        //        .build();
        //noKeyWindowResult.addSink(new OverSpeedAlarmSinkFunction(sinkProperties));
        env.execute("count-no-key-window");

    }


}

